#include "apictrl.h"
#include "apiconfig.h"

bool CAPICtrl::s_bReloadConfig = false;
bool CAPICtrl::s_bExit = false;
CAPICtrl* CAPICtrl::m_pstApiCtrl = NULL ;

void CAPICtrl::sigusr1_handle(int iSigVal)
{
    s_bReloadConfig = true;
    signal(SIGUSR1, CAPICtrl::sigusr1_handle);
}

void CAPICtrl::sigusr2_handle(int iSigVal)
{
    s_bExit = true;
    signal(SIGUSR2, CAPICtrl::sigusr2_handle);
}

CAPICtrl::CAPICtrl()
{
    m_bResetShm = true;

    // 统计项描述
    const char g_aszEnpStatDesc[][CTStatItem::CTLIB_DESC_MAX] = 
    {
        "RecvMessage",
        "SendMessage",
        "Accept",
        "IdleClose",
        "MainLoop",
        "WaitTimeout",
        "Check",
    };

    m_stStat.Init(CT_COUNTOF(g_aszEnpStatDesc), g_aszEnpStatDesc);
}

CAPICtrl::~CAPICtrl()
{
}

int CAPICtrl::Init()
{
    int iRet = 0;
    m_pstApiCtrl = this;

    //安装信号处理函数
    signal(SIGUSR1, CAPICtrl::sigusr1_handle);	// kill -USR1 pid 会让程序重读配置
    signal(SIGUSR2, CAPICtrl::sigusr2_handle);	// kill -USR2 pid 会让程序安全的退出

    // 读取配置
    iRet = CONF->LoadConfig();
    CHECK_THROW(iRet);

    // 创建Epoll
    iRet = m_stEpoll.Init(API_EVENT_SIZE, API_WAIT_SIZE);
    CHECK_THROW(iRet);

    // 最后创建ENP共享内存队列, 表示开始处理外部请求
    iRet = m_stEnpShmQueue.Init(true, 
        CONF->EnpFrontEndShmKey(), CONF->EnpFrontEndShmSize,
        CONF->EnpFrontEndSocket(), CONF->EnpBackEndSocket(),
        m_stEpoll,
        *this,
        &CAPICtrl::OnEnpShmQueueEvent);
    CHECK_THROW(iRet);

    iRet = m_stTApiShmQueue.Init(false, 
        CONF->TApiFrontEndShmKey(), CONF->TApiFrontEndShmSize,
        CONF->TApiFrontEndSocket(), CONF->TApiBackEndSocket(),
        m_stEpoll,
        *this,
        &CAPICtrl::OnTApiShmQueueEvent);
    CHECK_THROW(iRet);

    // 创建共享内存
#ifdef _DEBUG
    const char* pszConfig = "(512,3000)(1024,2000)(4096,1000)(16384,500)(65536,500)(300000,20)";
#else
    const char* pszConfig = "(512,6000)(1024,6000)(4096,6000)(16384,6000)(65536,4000)(131072,100)";
#endif

    const size_t iFactorySize = CT_CMDFACTORY->CountSize(pszConfig);

    const size_t iCountSize = sizeof(int) + iFactorySize;

    iRet = m_stMgrShm.Create(CONF->MgrShmKey(), iCountSize);
    CHECK_THROW(iRet);

    size_t BufOffset = 0;

    int* piOrgShmSize = (int*)(m_stMgrShm.GetBuffer() + BufOffset);
    if (!m_bResetShm && (size_t)*piOrgShmSize != iCountSize) //表示-u重启，需要恢复共享内存中的数据，但是共享内存中的长度和之前的不同
    {
        CT_WARNING(("shm size changed, -u param is invalid, %d, %d", *piOrgShmSize, iCountSize));
        m_bResetShm = true;
    }
    *piOrgShmSize = iCountSize;
    BufOffset += sizeof(int);

    // 初始化异步命令工厂
    iRet = CT_CMDFACTORY->Init(pszConfig, m_stMgrShm.GetBuffer() + BufOffset, iFactorySize);
    CHECK_THROW(iRet);
    BufOffset += iFactorySize;

    m_stDecodeBuf.Attach(NULL,
        CTMsgHead::EMAX_MSGLEN + sizeof(CTMsgHead) + sizeof(CEnpNetHead), 0);

    m_stEncodeBuf.Attach(NULL,
        CTMsgHead::EMAX_MSGLEN + sizeof(CTMsgHead) + sizeof(CEnpNetHead), 0);

    CT_INFO(("apisvr start"));
    return 0;
}

int CAPICtrl::Run()
{
    int iRet = 0;
    int iLoop = 0;

    m_uiNow = static_cast<unsigned int>(time(0));
    int iPrevNow = m_uiNow;

    while (true)
    {
        // 信号检查
        if (s_bReloadConfig)
        {
            s_bReloadConfig = false;
            CONF->ReloadConfig();
        }

        if (s_bExit)
        {
            s_bExit = false;
            CT_TRACE(("apisvr end!"));
            return 0;
        }

        m_stStat.Put(api_stat_mainloop);

        // 定时器检查
        if (iLoop == 0)
        {
            m_uiNow = (unsigned int) time(0);
            if (m_uiNow - iPrevNow >= 60)
            {
                CT_CMDFACTORY->CheckTimeoutCmd(m_uiNow);
                iPrevNow = m_uiNow;
                DumpStatInfo();
            }
        }

        iRet = m_stEpoll.WaitAndEvent(10);
        if (iRet == 0) // 超时了
        {
            m_stStat.Put(api_stat_waittimeout);
        }

        // 处理异步命令队列中命令
        iRet = CT_CMDFACTORY->DoQueue();
        CHECK_THROW(iRet);

        HandleEnpMsg();
        //HandleMpMsg();

        if (++iLoop >= 50)
        {
            iLoop = 0;
            m_stStat.Put(api_stat_check);     
        }
    }

    return 0;
}

int CAPICtrl::Exit()
{
    DumpStatInfo();
    CT_INFO(("%s exit", APPNAME));
    return 0;
}

void CAPICtrl::OnEnpShmQueueEvent(CAPIEpollObject* pstObject, SOCKET iSocket, int iEvent)
{
    CT_TRACE(("OnEnpShmQueueEvent socket=%d event=%d", iSocket, iEvent));
    m_stEnpShmQueue.ClearNotify(); // 接收Notify数据, 否则会一直收到事件
    //TODO: 每个消息都要执行ClearNotify, 可能存在性能问题, 以后验证

    if (!CTEpoll::IsInputEvent(iEvent))
    {
        CT_WARNING(("ShmQueue recv not match event(%d) socket=%d", iEvent, iSocket));
        return;
    }

    HandleEnpMsg();
}

void CAPICtrl::OnTApiShmQueueEvent(CAPIEpollObject* pstObject, SOCKET iSocket, int iEvent)
{
    CT_TRACE(("OnMpShmQueueEvent socket=%d event=%d", iSocket, iEvent));
    m_stTApiShmQueue.ClearNotify(); // 接收Notify数据, 否则会一直收到事件
    //TODO: 每个消息都要执行ClearNotify, 可能存在性能问题, 以后验证

    if (!CTEpoll::IsInputEvent(iEvent))
    {
        CT_WARNING(("ShmQueue recv not match event(%d) socket=%d", iEvent, iSocket));
        return;
    }

    HandleTApiMsg();
}

void CAPICtrl::DumpStatInfo()
{
    m_stStat.Dump(CT_STAT);

    //TODO: 还要加上其他的统计信息
	CT_CMDFACTORY->DumpStatInfo(CT_STAT);
	CT_CMDFACTORY->DumpStatDetail(CT_STAT);
}

void CAPICtrl::HandleEnpMsg()
{
    int iRet = 0;
    CTCodeStream s;
    CTMsgHead stMsgHead;

    for (int i = 0; i < 50; i++)  //此处的50为性能测试时通过反复调整获得的值
    {
        if (!m_stEnpShmQueue.HasCode())
        {
            return;
        }

        // 接收消息
        int iCodeLen = 0;
        m_stDecodeBuf.Clear();
        iRet = m_stEnpShmQueue.GetOneCode(m_stDecodeBuf.GetFreeBuf(), m_stDecodeBuf.GetFreeLen(), iCodeLen);
        if (iRet || iCodeLen < (int)sizeof(CEnpNetHead))
        {
            CT_WARNING(("get code from enpshmqueue failed ret=%d, codelen=%d", iRet, iCodeLen));
            continue;
        }
        m_stDecodeBuf.Append(iCodeLen);

        // 先获取NetHead
        CEnpNetHead& stHead = *(CEnpNetHead*)m_stDecodeBuf.GetUsedBuf();
        if (iCodeLen != (int)sizeof(CEnpNetHead) + stHead.unDataLength) // 长度不一致
        {
            CT_WARNING(("code length invalid. %d != %d + %d", iCodeLen,
                sizeof(CEnpNetHead), stHead.unDataLength));
            continue;
        }

        if (stHead.unDataLength <= 0)
        {
			CT_WARNING(("stHead.unDataLength <= 0"));
            continue;
        }

        CT_TRACE(("get one msg from enp, datalen(msghead+body) = %d", stHead.unDataLength));

        //TODO: 考虑增加IP限制的功能(原型阶段不实现)

        // 这里都是处理来自client的请求
        iRet = s.Attach(m_stDecodeBuf.GetUsedBuf() + sizeof(CEnpNetHead), stHead.unDataLength);
        CHECK_THROW(iRet);

        s.InitConvert();
        iRet = CodeConvert(s, stMsgHead, NULL, bin_decode());

        if (iRet != 0)
        {
            CT_ERROR(("Error in decoding message from enp, ret = %d", iRet));
            continue;
        }
        //CHECK_THROW(iRet);

        if (!stMsgHead.IsValid())
        {
            CT_WARNING(("invalid msghead, msgid = %d, msgtype = %d", stMsgHead.shMsgID, stMsgHead.shMsgType));
            continue;
        }

        // 创建异步命令
        CTAsyncCmd* pstCmd = (CTAsyncCmd*) CT_CMDFACTORY->CreateCmd(stMsgHead.shMsgID);
        if (!pstCmd)
        {
            CT_WARNING(("create cmd %d return null", stMsgHead.shMsgID));
            continue;
        }

        // 异步命令执行完后能自己释放
        CCmdDoParam stParam((void*)this, stHead, stMsgHead, s);
        iRet = pstCmd->Do(&stParam);
        CT_TRACE(("do cmd over, ret %d", iRet));
    }
}

int CAPICtrl::AnswerToClient(CEnpNetHead& stHead, CTCodeStream& s)
{
    CT_TRACE((""));
    int iRet = 0;
    stHead.unDataLength = s.m_ushPos;

    iRet = m_stEnpShmQueue.PutOneCode((const char*)&stHead, sizeof(stHead), s.m_pszBuf, s.m_ushPos);
    if (iRet)
    {
        CT_WARNING(("ret = %d", iRet));
        return iRet;
    }

	CT_TRACE(("send %d bytes to client", stHead.unDataLength));

    return iRet;
}

int CAPICtrl::AnswerToClient(CEnpNetHead& stHead, CTMsgHead& stMsgHead, CResult& stRet)
{
    CTCodeStream s(m_stEncodeBuf);
    CodeConvert(s, stMsgHead, NULL, bin_encode());
    CodeConvert(s, stRet, NULL, bin_encode());
    if (s.m_iRet)
    {
        CT_WARNING((""));
        return s.m_iRet;
    }

    return AnswerToClient(stHead, s);
}

int CAPICtrl::AnswerToClient(CEnpNetHead& stHead, const char* sAnsMsg, int MsgLen)
{
    int iRet = 0;
    stHead.unDataLength = MsgLen & 0xFFFF;
    iRet = m_stEnpShmQueue.PutOneCode((const char*)&stHead, sizeof(stHead), sAnsMsg, MsgLen);
    if (iRet)
    {
        CT_WARNING((""));
        return iRet;
    }

    return iRet;
}

int CAPICtrl::RequestToTApi(const char* pszBuf, int iBufLen)
{
    return m_stTApiShmQueue.PutOneCode(pszBuf, iBufLen);
}

void CAPICtrl::HandleTApiMsg()
{
    int iRet = 0;
    CTCodeStream s;
    CTMsgHead stMsgHead;

    for (;;)
    {
        if (!m_stTApiShmQueue.HasCode())
        {
            return;
        }

        // 接收消息
        int iCodeLen = 0;
        m_stDecodeBuf.Clear();
        iRet = m_stTApiShmQueue.GetOneCode(m_stDecodeBuf.GetFreeBuf(), m_stDecodeBuf.GetFreeLen(), iCodeLen);
        if (iRet || iCodeLen < 16) // CTMsgHead的编码长度是16
        {
            CT_WARNING(("get code from mpshmqueue failed ret=%d, codelen=%d", iRet, iCodeLen));            
            continue;
        }
        m_stDecodeBuf.Append(iCodeLen);

		// 先获取NetHead
		CEnpNetHead& stHead = *(CEnpNetHead*)m_stDecodeBuf.GetUsedBuf();
		if (iCodeLen != (int)sizeof(CEnpNetHead) + stHead.unDataLength) // 长度不一致
		{
			CT_WARNING(("code length invalid. %d != %d + %d", iCodeLen,
				sizeof(CEnpNetHead), stHead.unDataLength));
			continue;
		}

		if (stHead.unDataLength <= 0)
		{
			CT_WARNING(("stHead.unDataLength <= 0"));
			continue;
		}

		CT_TRACE(("get one msg from mp, datalen(msghead+body) = %d", stHead.unDataLength));

        // 这里都是处理来自mpsvr的请求
        iRet = s.Attach(m_stDecodeBuf.GetUsedBuf() + sizeof(CEnpNetHead), 
							(unsigned short)(m_stDecodeBuf.GetUsedLen() & 0xFFFF));
        CHECK_THROW(iRet);
		
        s.InitConvert();
        iRet = CodeConvert(s, stMsgHead, NULL, bin_decode());
        CHECK_THROW(iRet);

        if (!stMsgHead.IsValid())
        {
            CT_WARNING(("msghead invalid, stMsgHead.shMsgID = %d, stMsgHead.shMsgType = %d", 
							stMsgHead.shMsgID, stMsgHead.shMsgType));
            continue;
        }

        // 根据ACT找到对应发出请求的异步命令
        CTAsyncCmdInf* pstCmd = reinterpret_cast<CTAsyncCmdInf*>(stMsgHead.llMsgAct);
        if (!CT_CMDFACTORY->IsValidCmd(pstCmd))
        {
            CT_WARNING((""));
            continue;
        }

        // 检查消息的Seq是否匹配
        if (pstCmd->GetCmdSeq() != stMsgHead.iCmdSeq)
        {
            CT_WARNING((""));
            continue;
        }

        // 执行响应函数
        CCmdOnAnswerParam stParam((void*)this, stMsgHead, s);
        iRet = pstCmd->OnAnswer(&stParam);
    }
}



